package com.treasure.mall.biz.service.core.live.impl;

import com.treasure.mall.biz.base.AbstractSearchableCrudService;
import com.treasure.mall.biz.base.RedisLocker;
import com.treasure.mall.biz.base.RedisTemplate;
import com.treasure.mall.biz.base.SpecificationAndSort;
import com.treasure.mall.biz.config.redis.RedisConstant;
import com.treasure.mall.biz.dto.BaseSaveDTO;
import com.treasure.mall.biz.dto.BaseSearchDTO;
import com.treasure.mall.biz.dto.live.LiveRoomUserSaveDTO;
import com.treasure.mall.biz.dto.live.LiveSessionCondition;
import com.treasure.mall.biz.dto.live.LiveSessionCreateDTO;
import com.treasure.mall.biz.dto.live.SessionStatisticsDTO;
import com.treasure.mall.biz.entity.live.*;
import com.treasure.mall.biz.entity.user.UserPO;
import com.treasure.mall.biz.remote.live.TencentIMService;
import com.treasure.mall.biz.remote.live.TencentLiveService;
import com.treasure.mall.biz.remote.live.dto.GeneralMessageDTO;
import com.treasure.mall.biz.remote.live.enums.MessageType;
import com.treasure.mall.biz.repository.live.LiveSessionRepository;
import com.treasure.mall.biz.service.core.live.*;
import com.treasure.mall.biz.service.core.user.UserService;
import com.treasure.mall.zbq.common.Jsr310Utils;
import com.treasure.mall.zbq.common.RandomUtils;
import com.treasure.mall.zbq.common.SqlDataReader;
import com.treasure.mall.zbq.common.StringUtilsExt;
import com.treasure.mall.zbq.common.base.ApiResult;
import com.treasure.mall.zbq.common.base.ResultCodeEnum;
import com.treasure.mall.zbq.common.exception.ApiException;
import com.treasure.mall.zbq.common.ienum.live.LiveEnums;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import javax.persistence.criteria.Predicate;
import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * @author allan
 * @version 1.0.0
 * @date 2020/6/13
 */
@Service
@Slf4j
public class LiveSessionServiceImpl extends AbstractSearchableCrudService<LiveSessionPO, String> implements LiveSessionService {
    @Resource
    private LiveDirectorService directorService;
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private TencentLiveService tencentLiveService;
    @Resource
    private LiveRoomService liveRoomService;
    private LiveSessionRepository liveSessionRepository;

    @Autowired
    private ThreadPoolTaskScheduler scheduler;
    @Autowired
    private RedisLocker redisLocker;
    @Autowired
    private LiveRoomUserService liveRoomUserService;
    @Autowired
    private LiveRoomStatisticsService liveRoomStatisticsService;
    @Autowired
    private LiveRoomImagesService liveRoomImagesService;
    @Autowired
    private TencentIMService tencentIMService;
    @Autowired
    private UserService userService;

    @Autowired
    protected LiveSessionServiceImpl(LiveSessionRepository repository) {
        super(repository);
        this.liveSessionRepository = repository;
    }

    @Override
    protected SpecificationAndSort<LiveSessionPO> specificationAndSort(BaseSearchDTO searchDTO) {
        LiveSessionCondition condition = (LiveSessionCondition) searchDTO;

        Specification<LiveSessionPO> specification = (root, query, cb) -> {
            List<Predicate> predicates = new ArrayList<>();
            if (condition.getRoomId() > 0) {
                predicates.add(cb.equal(root.get(LiveSessionPO_.liveRoomPO).get(LiveRoomPO_.id), condition.getRoomId()));
            }

            return cb.and(predicates.toArray(new Predicate[predicates.size()]));
        };

        Sort sort = Sort.by(new Sort.Order(Sort.Direction.ASC, "playStatus"), new Sort.Order(Sort.Direction.DESC, "createTime"));

        return new SpecificationAndSort<>(specification, sort);
    }

    @Override
    @Transactional
    public void delete(String s) {
        LiveSessionPO sessionPO = findById(s);
        if (sessionPO.getPlayStatus() != LiveEnums.PlayStatus.TO_PLAY) {
            throw new ApiException(ApiResult.errorWith(ResultCodeEnum.ILLEGAL_STATE, "当前场次不能删除"));
        }
        repository.delete(sessionPO);
    }

    @Override
    @Transactional
    public LiveSessionPO save(BaseSaveDTO saveDTO) {
        LiveSessionCreateDTO createDTO = (LiveSessionCreateDTO) saveDTO;

        LiveSessionPO liveSessionPO;
        if (StringUtils.isEmpty(createDTO.getId())) {
            liveSessionPO = new LiveSessionPO();
            liveSessionPO.setId(RandomUtils.randomFlowNo(6));
        } else {
            liveSessionPO = findById(createDTO.getId());
        }

        liveSessionPO.setLiveDirectorPO(createDTO.getLiveDirector());
        liveSessionPO.setLiveRoomPO(new LiveRoomPO(createDTO.getLiveDirector().getRoomId()));
        liveSessionPO.setPlayStatus(LiveEnums.PlayStatus.TO_PLAY);
        if (!StringUtils.isEmpty(createDTO.getForeshowTime())) {
            liveSessionPO.setForeshowTime(Jsr310Utils.DateTime.of(createDTO.getForeshowTime()));
        }
        liveSessionPO.setForeshowDesc(createDTO.getForeshowDesc());
        liveSessionPO.setSessionTitle(createDTO.getSessionTitle());
        return save(liveSessionPO);
    }

    @Override
    public LiveSessionPO getCurrentSession(Long roomId) {
        LiveRoomPO roomPO = liveRoomService.findById(roomId);

        return roomPO.currentSession();
    }

    @Override
    @Transactional
    public LiveSessionPO create(LiveSessionCreateDTO createDTO) {
        return save(createDTO);
    }

    @Override
    @Transactional
    public LiveSessionPO start(Long roomId, String sessionTitle, Long userId) {
        //得到主持人
        LiveDirectorPO liveDirectorPO = directorService.findByRoomAndUser(roomId, userId);
        if (liveDirectorPO == null) {
            throw new ApiException(ApiResult.errorWith(ResultCodeEnum.LIVE_FAILED, "主持人无效"));
        }
        LiveRoomPO liveRoomPO = liveRoomService.findById(roomId);
        if (liveRoomPO.getAuditStatus() != LiveEnums.AuditStatus.SUCCESS) {
            throw new ApiException(ApiResult.errorWith(ResultCodeEnum.LIVE_FAILED, "该直播间未审核通过"));
        }
        if (!liveRoomPO.isEnabled()) {
            throw new ApiException(ApiResult.errorWith(ResultCodeEnum.LIVE_FAILED, "该直播间不可用"));
        }
        //找到最新的待开播场次
        LiveSessionPO liveSessionPO = liveSessionRepository.findTopByLiveRoomPO_IdAndPlayStatusOrderByForeshowTimeAsc(roomId, LiveEnums.PlayStatus.TO_PLAY);
        //如果没有创建一个场次
        if (liveSessionPO == null) {
            liveSessionPO = create(new LiveSessionCreateDTO(liveDirectorPO));
        }
        liveSessionPO.setSessionTitle(sessionTitle);

        //开始直播
        //得到推流地址
        ApiResult apiResult = tencentLiveService.setupPlayUrl(liveRoomPO);
        if (apiResult.isSuccess()) {
            //开启直播
            liveSessionPO.setPlayStatus(LiveEnums.PlayStatus.PAYING);
            liveSessionPO.setStartTime(LocalDateTime.now());
            liveSessionPO.setPlayUrl((String) apiResult.getData());

            liveSessionPO = save(liveSessionPO);
            //开启直播间状态
            liveRoomService.modifyLivePlaying(roomId, true);


            //进入房间
            userEnter(userId, liveSessionPO.getId());

            //更新用户在线总时长
            updateOnlineTime(userId, liveSessionPO.getId(), false);

            return liveSessionPO;
        }
        throw new ApiException(apiResult);
    }

    @Override
    public void stop(String sessionId) {
        LiveSessionPO liveSessionPO = findById(sessionId);
        stop(liveSessionPO);
    }

    @Override
    public void stop(LiveSessionPO liveSessionPO) {
        if (liveSessionPO.getPlayStatus() == LiveEnums.PlayStatus.PAY_COMPLETE) {
            throw new ApiException(ApiResult.errorWith(ResultCodeEnum.ILLEGAL_STATE, "直播已结束"));
        }
        liveSessionPO.setPlayStatus(LiveEnums.PlayStatus.PAY_COMPLETE);
        liveSessionPO.setEndTime(LocalDateTime.now());
        //停止机器人
        liveSessionPO.setEnableRobot(false);

        //那么一些统计数据
        SessionStatisticsDTO statisticsDTO = this.getStatistics(liveSessionPO.getId());
        liveSessionPO.setMaxOnlineNum(statisticsDTO.getMaxOnlineNum());
        liveSessionPO.setCurrentMaxVisits(statisticsDTO.getGrandTotalNum());
        liveSessionPO.setDealNum(statisticsDTO.getDealNum());
        liveSessionPO.setDealAmount(statisticsDTO.getDealAmount());
        liveSessionPO.setActivityNum(statisticsDTO.getActivityNum());
        liveSessionPO.setUserOnlineTotalSeconds(statisticsDTO.getUserOnlineTotalSeconds());


        //更新用户在线总时长
        updateOnlineTime(liveSessionPO.getLiveDirectorPO().getUserPO().getUserId(), liveSessionPO.getId(), true);
        //离开直播间
        userLeave(liveSessionPO.getLiveDirectorPO().getUserPO().getUserId(), liveSessionPO.getId());

        //删除redis中的数据
        Set<String> keys = redisTemplate.keys(RedisConstant.KEY_LIVE_SESSION + liveSessionPO.getId() + "*");
        redisTemplate.delete(keys);


        save(liveSessionPO);
        //关闭相关服务
        stopOtherService(liveSessionPO);

    }

    /**
     * 直播间关闭后，需要关闭相对应的服务
     *
     * @param liveSessionPO
     */
    private void stopOtherService(LiveSessionPO liveSessionPO) {
        try {
            //关闭直播间状态
            liveRoomService.modifyLivePlaying(liveSessionPO.getLiveRoomPO().getId(), false);
            //关闭客服通知
            liveRoomService.enableNotice(liveSessionPO.getLiveRoomPO().getId(), false);


        } catch (Exception e) {

        }

    }


    @Override
    @Transactional
    public void stopByRoom(Long roomId) {
        List<LiveSessionPO> liveSessionPOS = findByRoom(roomId);
        for (LiveSessionPO liveSessionPO : liveSessionPOS) {
            stop(liveSessionPO);
        }
    }

    /**
     * 进入直播间
     *
     * @param userId
     * @param sessionId
     */
    @Override
    public void userEnter(Long userId, String sessionId) {
        String memKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":members";
        String memHashKey = "mem_" + userId;
        //在线添加
        redisTemplate.opsForHash().putIfAbsent(memKey + RedisConstant.HASH_KEY_USER_ONLINE, memHashKey, userId);
        //离开的删除
        redisTemplate.opsForHash().delete(memKey + RedisConstant.HASH_KEY_USER_OFFLINE, memHashKey);

        //增加在线人数
        String statisticsKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":statistics";
        long onlineNum = redisTemplate.opsForHash().increment(statisticsKey, RedisConstant.HASH_KEY_ONLINE_NUM, 1);

        //最高在线人数
        if (redisTemplate.opsForHash().hasKey(statisticsKey, RedisConstant.HASH_KEY_MAX_ONLINE)) {
            int maxOnlineNum = (int) redisTemplate.opsForHash().get(statisticsKey, RedisConstant.HASH_KEY_MAX_ONLINE);
            if (onlineNum > maxOnlineNum) {
                //覆盖最高
                redisTemplate.opsForHash().put(statisticsKey, RedisConstant.HASH_KEY_MAX_ONLINE, onlineNum);
            }
        } else {
            redisTemplate.opsForHash().put(statisticsKey, RedisConstant.HASH_KEY_MAX_ONLINE, onlineNum);
        }
        //活跃度
        redisTemplate.opsForHash().increment(statisticsKey, RedisConstant.HASH_KEY_ACTIVITY_NUM, 1);


        /**
         * 设置用户在线记录
         */
        LiveRoomUserSaveDTO userSaveDTO = new LiveRoomUserSaveDTO();
        userSaveDTO.setSessionId(sessionId);
        userSaveDTO.setUserId(userId);
        liveRoomUserService.save(userSaveDTO);
    }

    /**
     * 添加活跃度
     *
     * @param sessionId
     */
    @Override
    public void addActivityNum(String sessionId) {
        addActivityNum(sessionId, 1);
    }

    @Override
    public void addActivityNum(String sessionId, int num) {
        String statisticsKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":statistics";
        redisTemplate.opsForHash().increment(statisticsKey, RedisConstant.HASH_KEY_ACTIVITY_NUM, num);
    }


    @Override
    public void userLeave(Long userId, String sessionId) {
        String memKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":members";
        String memHashKey = "mem_" + userId;
        //在线的删除
        redisTemplate.opsForHash().delete(memKey + RedisConstant.HASH_KEY_USER_ONLINE, memHashKey);
        //离开的增加
        redisTemplate.opsForHash().putIfAbsent(memKey + RedisConstant.HASH_KEY_USER_OFFLINE, memHashKey, userId);

        //减少在线人数
        String statisticsKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":statistics";
        redisTemplate.opsForHash().increment(statisticsKey, RedisConstant.HASH_KEY_ONLINE_NUM, -1);


        //设置用户离线状态
        liveRoomUserService.modifyUserOffline(userId, sessionId);

    }

    /**
     * 更新在线时间
     *
     * @param userId    用户ID
     * @param sessionId sessionId
     * @param isLeave   是否是离开
     */
    @Override
    public void updateOnlineTime(Long userId, String sessionId, boolean isLeave) {
        while (true) {
            String lockKey = RedisConstant.LOCK_KEY_UPDATE_ONLINE_TIME + userId;
            boolean locker = redisLocker.tryLock(lockKey, RedisConstant.LOCK_KEY_UPDATE_ONLINE_TIME, 10);
            if (locker) {
                try {
                    String memKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":members";
                    String memHashKey = "mem_" + userId;
                    String statisticsTimeKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":statistics:time";
                    long totalSeconds = 0;
                    //获取上线时间
                    LocalDateTime onlineDateTime = redisTemplate.getLocalDateTime(memKey + RedisConstant.HASH_KEY_USER_ONLINE_TIME, memHashKey);
                    if (onlineDateTime != null) {
                        //获取当前时长
                        Duration duration = Duration.between(onlineDateTime, LocalDateTime.now());
                        totalSeconds = duration.getSeconds();
                        //设置用户在线时长
                        liveRoomUserService.modifyUserOnlineDuration(userId, sessionId, totalSeconds);
                    }
                    //计算当前用户累计在线总时间
                    if (redisTemplate.opsForHash().hasKey(statisticsTimeKey, RedisConstant.HASH_KEY_TOTAL_ONLINE_TIME)) {
                        totalSeconds += redisTemplate.getLong(statisticsTimeKey, RedisConstant.HASH_KEY_TOTAL_ONLINE_TIME);
                    }
                    //更新用户在线总时长(所有的用户)
                    redisTemplate.opsForHash().put(statisticsTimeKey, RedisConstant.HASH_KEY_TOTAL_ONLINE_TIME, totalSeconds);
                    //删除上线时间
                    redisTemplate.opsForHash().delete(memKey + RedisConstant.HASH_KEY_USER_ONLINE_TIME, memHashKey);
                    //判断是否是离开，否则需要重新更新上线时间
                    if (!isLeave) {
                        redisTemplate.opsForHash().putIfAbsent(memKey + RedisConstant.HASH_KEY_USER_ONLINE_TIME, memHashKey, Jsr310Utils.DateTime.asString(LocalDateTime.now()));
                        liveRoomUserService.modifyUserOnline(userId, sessionId);
                    }
                } finally {
                    redisLocker.release(lockKey, RedisConstant.LOCK_KEY_UPDATE_ONLINE_TIME);
                }
                break;
            }
        }
    }


    @Override
    public void deal(String sessionId, BigDecimal amount) {
        String statisticsKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":statistics";
        //增加订单数量
        redisTemplate.opsForHash().increment(statisticsKey, RedisConstant.HASH_KEY_DEAL_NUM, 1);
        //增加成交金额
        redisTemplate.opsForHash().increment(statisticsKey, RedisConstant.HASH_KEY_DEAL_AMOUNT, amount.doubleValue());
    }

    @Override
    public SessionStatisticsDTO getStatistics(String sessionId) {
        String statisticsKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":statistics";
        String memKey = RedisConstant.KEY_LIVE_SESSION + sessionId + ":members";
        String statisticsTimeKey = statisticsKey + ":time";

        int dealNum = redisTemplate.getInt(statisticsKey, RedisConstant.HASH_KEY_DEAL_NUM);
        BigDecimal dealAmount = redisTemplate.getBigDecimal(statisticsKey, RedisConstant.HASH_KEY_DEAL_AMOUNT, 2);
        long onlineNum = redisTemplate.opsForHash().size(memKey + RedisConstant.HASH_KEY_USER_ONLINE);
        long offlineNum = redisTemplate.opsForHash().size(memKey + RedisConstant.HASH_KEY_USER_OFFLINE);
        long grandTotalNum = onlineNum + offlineNum;
        int maxOnlineNum = redisTemplate.getInt(statisticsKey, RedisConstant.HASH_KEY_MAX_ONLINE);
        int activityNum = redisTemplate.getInt(statisticsKey, RedisConstant.HASH_KEY_ACTIVITY_NUM);

        /**
         * 用户在线总时长
         */
        long userOnlineTotalSeconds = redisTemplate.getLong(statisticsTimeKey, RedisConstant.HASH_KEY_TOTAL_ONLINE_TIME);

        SessionStatisticsDTO statisticsDTO = new SessionStatisticsDTO();
        statisticsDTO.setDealAmount(dealAmount);
        statisticsDTO.setOnlineNum(onlineNum);
        statisticsDTO.setDealNum(dealNum);
        statisticsDTO.setGrandTotalNum(grandTotalNum);
        statisticsDTO.setMaxOnlineNum(maxOnlineNum);
        statisticsDTO.setActivityNum(activityNum);
        statisticsDTO.setUserOnlineTotalSeconds(userOnlineTotalSeconds);

        return statisticsDTO;
    }

    @Override
    public List<LiveSessionPO> findByRoom(Long roomId) {
        return liveSessionRepository.findByLiveRoomPO_IdAndPlayStatus(roomId, LiveEnums.PlayStatus.PAYING);
    }

    /**
     * 开启机器人
     *
     * @param sessionId
     * @param enable
     */
    @Override
    public void enableRobot(String sessionId, Boolean enable) {
        liveSessionRepository.enableRobot(sessionId, enable);
    }


    /**
     * 检查直播间是否已经断流
     */
    @Override
    public void checkLiveState(long seconds) {
        try {
            List<LiveSessionPO> pos = liveSessionRepository.findByPlayStatus(LiveEnums.PlayStatus.PAYING);
            if (pos != null && pos.size() > 0) {
                pos.forEach(item -> {
                    //检查是否是激活状态
                    boolean state = tencentLiveService.describeLiveStreamState(item.getLiveRoomPO().toStreamName());
                    if (!state) {
                        scheduler.schedule(() -> stopLiveState(item.getId()), Jsr310Utils.DateTime.toDate(LocalDateTime.now().plusSeconds(seconds)));
                    }
                });
            }
        } catch (Exception e) {
            log.error("checkLiveState", e);
        }
    }


    public void stopLiveState(String liveSessionId) {
        try {
            LiveSessionPO po = findById(liveSessionId);
            if (po == null || !po.getPlayStatus().equals(LiveEnums.PlayStatus.PAYING)) {
                return;
            }
            boolean state = tencentLiveService.describeLiveStreamState(po.getLiveRoomPO().toStreamName());
            if (!state) {
                stop(liveSessionId);
            }
        } catch (Exception e) {
            log.error("自动关闭不活动的直播间", e);
        }
    }


    /**
     * 添加进入直播间的方式
     *
     * @param userId
     * @param sessionId
     * @param scene
     */
    @Override
    public void addEnterRoomWay(Long userId, String sessionId, String scene) {
        if (StringUtils.isEmpty(scene)) {
            return;
        }
        liveRoomStatisticsService.add(userId, sessionId, scene);
    }

    /**
     * 发送直播间消息
     *
     * @param sessionId
     * @param userId    用户ID
     */
    @Override
    public void sendRoomMessage(String sessionId, Long userId) {
        LiveSessionPO liveSessionPO = findById(sessionId);
        if (liveSessionPO == null) {
            return;
        }

        //直播间消息必须判断用户是否存在
        UserPO userPO = userService.findById(userId);
        if (userPO == null) {
            return;
        }


        LiveRoomPO liveRoomPO = liveSessionPO.getLiveRoomPO();
        //检查当前是否已经开启公告
        if (liveRoomPO.isEnableNotice() && !StringUtils.isEmpty(liveRoomPO.getNotice())) {
            // 发送消息
            //发送公告消息
            tencentIMService.sendMessage(GeneralMessageDTO.builder()
                    .roomId(liveRoomPO.getId())
                    .message(StringUtilsExt.replaceEnterStr(liveRoomPO.getNotice()))
                    .userId(userId)
                    .build(), MessageType.notice);
        }


        //发送直播间图片消息
        liveRoomImagesService.sendRoomImageMessage(liveSessionPO.getLiveRoomPO().getId(), userId);
    }

    /**
     * 检查当前用户是否已经在直播中，用户验证码登录时检查
     *
     * @param userId
     * @return
     */
    @Override
    public boolean checkPlaying(Long userId) {
        String sql = "SELECT COUNT(1) as num from zb_live_session a LEFT JOIN zb_live_director b on a.director_id=b.id where b.rel_user_id= " + userId + " and a.play_status=0";
        SqlDataReader reader = executeReader(sql);
        if (reader != null) {
            return reader.getInteger("num") > 0;
        }
        return false;
    }

}
